package com.yht.redbook.note.biz.consumer;

import com.alibaba.nacos.shaded.com.google.common.util.concurrent.RateLimiter;
import com.yht.framework.common.utils.JsonUtils;
import com.yht.redbook.note.biz.constant.MQConstants;
import com.yht.redbook.note.biz.domain.dataobject.NoteCollectionDO;
import com.yht.redbook.note.biz.domain.mapper.NoteCollectionDOMapper;
import com.yht.redbook.note.biz.model.dto.CollectUnCollectNoteDTO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.Objects;

/**
 * @author: yht
 * @date: 2025/2/23 11:22
 * @description:
 **/

@RocketMQMessageListener(consumerGroup = "redbook_group_" + MQConstants.TOPIC_COLLECT_OR_UN_COLLECT,
        topic = MQConstants.TOPIC_COLLECT_OR_UN_COLLECT, consumeMode = ConsumeMode.ORDERLY)
@Slf4j
@Component
public class CollectUnCollectNoteConsumer implements RocketMQListener<Message> {

    @Resource
    private NoteCollectionDOMapper noteCollectionDOMapper;

    // 每秒创建 5000 个令牌
    private RateLimiter rateLimiter = RateLimiter.create(5000);

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Override
    public void onMessage(Message message) {
        rateLimiter.acquire();

        // 幂等性：通过联合唯一索引保证
        // 消息体
        String bodyJsonStr = new String(message.getBody());

        // 消息标签
        String tags = message.getTags();

        log.info("==> CollectUnCollectNoteConsumer 消费了消息 {}, tags: {}", bodyJsonStr, tags);

        // 根据 MQ 标签，判断操作类型
        if (Objects.equals(tags, MQConstants.TAG_COLLECT)) {
            handleConllectNoteTagMessage(bodyJsonStr);
        } else if (Objects.equals(tags, MQConstants.TAG_UN_COLLECT)) {
            handleUnConllectNoteTagMessage(bodyJsonStr);
        }
    }

    /**
     * 收藏笔记
     *
     * @param bodyJsonStr
     */
    private void handleConllectNoteTagMessage(String bodyJsonStr) {
        CollectUnCollectNoteDTO collectUnCollectNoteDTO = JsonUtils.parseObject(bodyJsonStr, CollectUnCollectNoteDTO.class);

        if (Objects.isNull(collectUnCollectNoteDTO)) return;

        Long noteId = collectUnCollectNoteDTO.getNoteId();
        Long userId = collectUnCollectNoteDTO.getUserId();

        LocalDateTime createTime = collectUnCollectNoteDTO.getCreateTime();
        Integer type = collectUnCollectNoteDTO.getType();

        NoteCollectionDO noteCollectionDO = NoteCollectionDO.builder()
                .noteId(noteId)
                .userId(userId)
                .createTime(createTime)
                .status(type)
                .build();

        int count = noteCollectionDOMapper.insertOrUpdate(noteCollectionDO);

        // 发送 MQ 计数
        if (count == 0) return;

        // 更新数据库成功后，发送计数 MQ
        org.springframework.messaging.Message<String> message = MessageBuilder.withPayload(bodyJsonStr).build();

        // 异步发送 MQ 消息
        rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_NOTE_COLLECT, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【笔记收藏】 MQ 发送成功，SendResult：{}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.info("==> 【笔记收藏】 MQ 发送异常，", throwable);
            }
        });
    }

    /**
     * 取消收藏
     *
     * @param bodyJsonStr
     */
    private void handleUnConllectNoteTagMessage(String bodyJsonStr) {
        CollectUnCollectNoteDTO collectUnCollectNoteDTO = JsonUtils.parseObject(bodyJsonStr, CollectUnCollectNoteDTO.class);

        if (Objects.isNull(collectUnCollectNoteDTO)) return;

        Long userId = collectUnCollectNoteDTO.getUserId();
        Long noteId = collectUnCollectNoteDTO.getNoteId();
        Integer type = collectUnCollectNoteDTO.getType();
        LocalDateTime createTime = collectUnCollectNoteDTO.getCreateTime();

        NoteCollectionDO noteCollectionDO = NoteCollectionDO.builder()
                .noteId(noteId)
                .userId(userId)
                .status(type)
                .createTime(createTime)
                .build();

        int count = noteCollectionDOMapper.update2UnCollectByUserIdAndNoteId(noteCollectionDO);

        // 发送 MQ 计数
        if (count == 0) return;

        // 更新数据库成功后，发送计数 MQ
        org.springframework.messaging.Message<String> message = MessageBuilder.withPayload(bodyJsonStr).build();

        // 异步发送 MQ 消息
        rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_NOTE_COLLECT, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【笔记收藏】 MQ 发送成功，SendResult：{}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.info("==> 【笔记收藏】 MQ 发送异常，", throwable);
            }
        });
    }
}
